package cn.lg.monitor

import com.lg.bean.BusInfo
import org.apache.spark.sql.{ForeachWriter, SparkSession}
import redis.clients.jedis.{JedisPool, JedisPoolConfig}

import java.sql.DriverManager

object MySqlWriter {

}

class MySqlWriter  extends ForeachWriter[BusInfo] {

  //开启连接
  override def open(partitionId: Long, epochId: Long): Boolean = {
    //如果表不存在就创建新表
    val sql = "create table if not exists test.bus_info(deployNum VARCHAR(100), lglat VARCHAR(100))"
    Class.forName("com.mysql.jdbc.Driver")
    val connection = DriverManager.getConnection("jdbc:mysql://linux123:3306/test? useSSL=false&characterEncoding=utf8","root","12345678")
    connection.createStatement().execute(sql)
    true
  }
  //处理数据
  override def process(value: BusInfo): Unit = {
    //把数据写入redis ,kv形式

    val lglat: String = value.lglat
    val deployNum = value.deployNum
    val sql = "insert into test.bus_info(deployNum ,lglat) values ('"+ deployNum +"','"+lglat+"')"
    Class.forName("com.mysql.jdbc.Driver")
    val connection = DriverManager.getConnection("jdbc:mysql://linux123:3306/test? useSSL=false&characterEncoding=utf8","root","12345678")
    connection.createStatement().execute(sql)
  }

  //释放连接
  override def close(errorOrNull: Throwable): Unit = {
//    connection.close()
  }

}
